package org.apache.rocketmq.namesrv.processor;

import io.netty.channel.ChannelHandlerContext;
import org.apache.rocketmq.common.MQVersion;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.UtilAll;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.namesrv.NamesrvController;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingCommandException;
import org.apache.rocketmq.remoting.netty.NettyRequestProcessor;
import org.apache.rocketmq.remoting.protocol.*;
import org.apache.rocketmq.remoting.protocol.body.RegisterBrokerBody;
import org.apache.rocketmq.remoting.protocol.body.TopicConfigSerializeWrapper;
import org.apache.rocketmq.remoting.protocol.header.DeleteTopicFromNamesrvRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.namesrv.*;
import org.apache.rocketmq.remoting.protocol.namesrv.RegisterBrokerResult;
import org.apache.rocketmq.remoting.protocol.namesrv.RegisterTopicRequestHeader;
import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;

import java.io.IOException;
import java.util.Arrays;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.atomic.AtomicLong;


/**
 * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
 * @author：陈清风扬，个人微信号：chenqingfengyangjj。
 * @date:2024/12/27
 * @方法描述：处理默认请求的处理器，这个请求处理器会处理除了客户端发送的根据主题获取路由信息的请求之外的所有请求
 */
public class DefaultRequestProcessor implements NettyRequestProcessor {

    private static Logger log = LoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);


    protected final NamesrvController namesrvController;


    protected Set<String> configBlackList = new HashSet<>();


    public DefaultRequestProcessor(NamesrvController namesrvController) {
        this.namesrvController = namesrvController;
        initConfigBlackList();
    }

    //初始化配置黑名单的方法，所谓配置黑名单，就是一些不允许客户端修改的配置信息，比如namesrv的配置文件路径，rocketmq的home路径等等
    private void initConfigBlackList() {
        configBlackList.add("configBlackList");
        configBlackList.add("configStorePath");
        configBlackList.add("kvConfigPath");
        configBlackList.add("rocketmqHome");
        String[] configArray = namesrvController.getNamesrvConfig().getConfigBlackList().split(";");
        configBlackList.addAll(Arrays.asList(configArray));
    }



    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/7
     * @方法描述：该方法就是默认请求处理器处理请求的方法，实际上，在Namesrv模块中，只有根据主题获得对应路由信息的请求会交给ClientRequestProcessor客户端请求处理器处理
     * 其余接收到的所有请求都会交给这个默认处理器处理
     */
    @Override
    public RemotingCommand processRequest(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
        //判断请求上下文对象是否不为空，不为空则记录请求日志
        if (ctx != null) {
            log.debug("receive request, {} {} {}",
                    //记录请求类型
                    request.getCode(),
                    //记录发送请求过来的客户端的地址
                    RemotingHelper.parseChannelRemoteAddr(ctx.channel()),
                    //记录请求本身的内容
                    request);
        }//判断请求类型
        switch (request.getCode()) {
            //处理Broker客户端发送过来的注册Broker信息的请求
            case RequestCode.REGISTER_BROKER:
                return this.registerBroker(ctx, request);

            case RequestCode.UNREGISTER_BROKER:
                return this.unregisterBroker(ctx, request);
            //处理Broker客户端发送过来的注销Broker信息的请求
            case RequestCode.REGISTER_TOPIC_IN_NAMESRV:
                return this.registerTopicToNamesrv(ctx, request);
            case RequestCode.GET_BROKER_CLUSTER_INFO:
                return this.getBrokerClusterInfo(ctx, request);
            //处理查询数据版本的请求
            case RequestCode.QUERY_DATA_VERSION:
                return this.queryBrokerTopicConfig(ctx, request);
            //处理删除主题的请求
            case RequestCode.DELETE_TOPIC_IN_NAMESRV:
                return this.deleteTopicInNamesrv(ctx, request);
            default:
                String error = " request type " + request.getCode() + " not supported";
                return RemotingCommand.createResponseCommand(RemotingSysResponseCode.REQUEST_CODE_NOT_SUPPORTED, error);
        }
    }

    private RemotingCommand deleteTopicInNamesrv(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
        final RemotingCommand response = RemotingCommand.createResponseCommand(null);
        final DeleteTopicFromNamesrvRequestHeader requestHeader = (DeleteTopicFromNamesrvRequestHeader) request.decodeCommandCustomHeader(DeleteTopicFromNamesrvRequestHeader.class);
        if (requestHeader.getClusterName() != null && !requestHeader.getClusterName().isEmpty()) {
            //根据集群名称和主题名称删除主题
            this.namesrvController.getRouteInfoManager().deleteTopic(requestHeader.getTopic(), requestHeader.getClusterName());
        } else {
            //根据主题名称删除主题
            this.namesrvController.getRouteInfoManager().deleteTopic(requestHeader.getTopic());
        }
        response.setCode(ResponseCode.SUCCESS);
        response.setRemark(null);
        return response;
    }

    public RemotingCommand queryBrokerTopicConfig(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {

        //创建响应对象
        final RemotingCommand response = RemotingCommand.createResponseCommand(QueryDataVersionResponseHeader.class);
        //得到响应头
        final QueryDataVersionResponseHeader responseHeader = (QueryDataVersionResponseHeader) response.readCustomHeader();
        //解码请求，得到请求头
        final QueryDataVersionRequestHeader requestHeader =
                (QueryDataVersionRequestHeader) request.decodeCommandCustomHeader(QueryDataVersionRequestHeader.class);
        //解码请求体，得到请求体中Broker发送过来的主题配置信息的数据版本
        DataVersion dataVersion = DataVersion.decode(request.getBody(), DataVersion.class);
        //从请求头中得到集群名称
        String clusterName = requestHeader.getClusterName();
        //从请求头中得到Broker节点地址
        String brokerAddr = requestHeader.getBrokerAddr();
        //从路由信息管理器中查询，Namesrv中存储的主题配置信息数据版本是否和发送请求过来的Broker主题配置信息数据版本不一样
        Boolean changed = this.namesrvController.getRouteInfoManager().isBrokerTopicConfigChanged(clusterName, brokerAddr, dataVersion);
        //在这里更新Namesrv中存储的Broker信息的最新时间戳，看到这里大家就会意识到，每一次Broker向Namesrv发送QueryDataVersionResponseHeader请求时候，都会更新Namesrv中存储的Broker信息的最新时间戳
        //而Broker会定期向Namesrv发送QueryDataVersionResponseHeader请求，这样一来，Namesrv中存储的Broker信息的时间戳就会一直保持最新，Namesrv中存储的Broker信息就不会过期，也就不会被Namesrv因为心跳检测失败而移除了
        this.namesrvController.getRouteInfoManager().updateBrokerInfoUpdateTimestamp(clusterName, brokerAddr);
        //查询Namesrv中存储的主题配置信息数据版本
        DataVersion nameSeverDataVersion = this.namesrvController.getRouteInfoManager().queryBrokerTopicConfig(clusterName, brokerAddr);
        //设置成功响应
        response.setCode(ResponseCode.SUCCESS);
        response.setRemark(null);
        if (nameSeverDataVersion != null) {
            //如果Namesrv中存储的主题配置信息数据版本不为空，那么就把这个数据版本编码后设置到响应体中
            response.setBody(nameSeverDataVersion.encode());
        }
        //把Namesrv中存储的主题配置信息数据版本和Broker的数据版本是否一致的标志设置到响应头中
        responseHeader.setChanged(changed);
        return response;
    }
    private RemotingCommand registerTopicToNamesrv(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {
        //创建响应对象
        final RemotingCommand response = RemotingCommand.createResponseCommand(null);
        //解码请求，得到请求头对象
        final RegisterTopicRequestHeader requestHeader = (RegisterTopicRequestHeader) request.decodeCommandCustomHeader(RegisterTopicRequestHeader.class);
        //解码请求体，得到主题路由数据对象
        TopicRouteData topicRouteData = TopicRouteData.decode(request.getBody(), TopicRouteData.class);
        //如果主题路由数据对象不为空，且主题路由数据对象中的队列数据不为空且不为空集合
        if (topicRouteData != null && topicRouteData.getQueueDatas() != null && !topicRouteData.getQueueDatas().isEmpty()) {
            //在这里把主题路由数据注册到路由信息管理器中
            this.namesrvController.getRouteInfoManager().registerTopic(requestHeader.getTopic(), topicRouteData.getQueueDatas());
        }
        //设置成功响应
        response.setCode(ResponseCode.SUCCESS);
        response.setRemark(null);
        return response;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/8
     * @方法描述：注册Broker信息到Namesrv的路由管理器中的方法
     */
    public RemotingCommand registerBroker(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {

        //首先创建一个响应对象
        final RemotingCommand response = RemotingCommand.createResponseCommand(RegisterBrokerResponseHeader.class);
        //创建响应头对象
        final RegisterBrokerResponseHeader responseHeader = (RegisterBrokerResponseHeader) response.readCustomHeader();
        //解码请求头对象，得到请求头
        final RegisterBrokerRequestHeader requestHeader = (RegisterBrokerRequestHeader) request.decodeCommandCustomHeader(RegisterBrokerRequestHeader.class);
        //检查请求体的CRC32校验和是否和请求头中的CRC32校验和匹配
        if (!checksum(ctx, request, requestHeader)) {
            //如果校验和不匹配，意味着数据可能在网络传输中丢失了一部分，不完整了，这个时候直接返回系统错误响应即可
            response.setCode(ResponseCode.SYSTEM_ERROR);
            response.setRemark("crc32 not match");
            return response;
        }
        TopicConfigSerializeWrapper topicConfigWrapper = null;
        List<String> filterServerList = null;
        MQVersion.Version brokerVersion = MQVersion.value2Version(request.getVersion());
        if (brokerVersion.ordinal() >= MQVersion.Version.V3_0_11.ordinal()){
            final RegisterBrokerBody registerBrokerBody = extractRegisterBodyFromRequest(request,requestHeader);
            //得到主题配置信息对象
            topicConfigWrapper = registerBrokerBody.getTopicConfigSerializeWrapper();
            filterServerList = registerBrokerBody.getFilterServerList();
        } else {
            //如果为3.0.11版本以下的Broker，那么就从请求中得到封装了主题配置信息的对象
            topicConfigWrapper = extractRegisterTopicConfigFromRequest(request);
        }
        log.info("receive RegisterBroker body {}", topicConfigWrapper);
        //以下这行代码是我自己添加的，用来验证测试类的正确性
        //从请求头中得到Broker的各个信息，然后把这些信息注册到路由信息管理器中
        RegisterBrokerResult result = this.namesrvController.getRouteInfoManager().registerBroker(
                //得到Broker所属的集群名称
                requestHeader.getClusterName(),
                //得到Broker的IP地址
                requestHeader.getBrokerAddr(),
                //得到Broker的名称
                requestHeader.getBrokerName(),
                //得到Broker的ID
                requestHeader.getBrokerId(),
                //得到Broker的高可用地址
                requestHeader.getHaServerAddr(),
                //得到区域名称
                request.getExtFields().get(MixAll.ZONE_NAME),
                //得到心跳超时时间
                requestHeader.getHeartbeatTimeoutMillis(),
                //得到是否启用ActingMaster模式的标志
                requestHeader.getEnableActingMaster(),
                //得到Topic配置
                topicConfigWrapper,
                null,
                //把当前发送请求的Broker和Namesrv构建的channel通道传进去
                ctx.channel()
        );
        //这里得到了注册结果，如果注册结果为空，说明注册操作失败了
        if (result == null) {
            //直接返回客户端错误响应即可
            response.setCode(ResponseCode.SYSTEM_ERROR);
            response.setRemark("register broker failed");
            return response;
        }
        //执行到这里意味着注册操作执行成功，这里把当前Broker所属集群的高可用地址和主节点地址设置到响应头中设置到响应头中
        responseHeader.setHaServerAddr(result.getHaServerAddr());
        //设置集群主节点地址，这里我多解释一下，因为在Broker集群中，从节点也会把自己的信息中注册到Namesrv中，从节点注册自己信息的时候
        //Namesrv会把从节点所属节点的主节点地址以及高可用地址告诉从节点，这样一来从节点就可以根据这个地址和集群主节点建立连接，然后同步数据了
        //当然，在第五版本代码中根本就没有引入Broker集群，我这里之所以没有把和集群相关的代码省略了，是想省点事，不然后面要对代码进行重构的地方就太多了
        //我怕我记不住，导致程序出现一些本来可以避免的bug
        responseHeader.setMasterAddr(result.getMasterAddr());
        //设置成功响应
        response.setCode(ResponseCode.SUCCESS);
        response.setRemark(null);
        return response;
    }

    private TopicConfigSerializeWrapper extractRegisterTopicConfigFromRequest(RemotingCommand request) {
        TopicConfigSerializeWrapper topicConfigWrapper;
        if (request.getBody() != null) {
            topicConfigWrapper = TopicConfigSerializeWrapper.decode(request.getBody(), TopicConfigSerializeWrapper.class);
        } else {
            topicConfigWrapper = new TopicConfigSerializeWrapper();
            topicConfigWrapper.getDataVersion().setCounter(new AtomicLong(0));
            topicConfigWrapper.getDataVersion().setTimestamp(0L);
            topicConfigWrapper.getDataVersion().setStateVersion(0L);
        }
        return topicConfigWrapper;
    }

    private RegisterBrokerBody extractRegisterBodyFromRequest(RemotingCommand request, RegisterBrokerRequestHeader requestHeader) throws RemotingCommandException {
        RegisterBrokerBody registerBrokerBody = new RegisterBrokerBody();
        if (request.getBody() != null) {
            try {
                MQVersion.Version brokerVersion = MQVersion.value2Version(request.getVersion());
                registerBrokerBody = RegisterBrokerBody.decode(request.getBody(), requestHeader.isCompressed(), brokerVersion);
            } catch (Exception e) {
                throw new RemotingCommandException("Failed to decode RegisterBrokerBody", e);
            }
        } else {
            registerBrokerBody.getTopicConfigSerializeWrapper().getDataVersion().setCounter(new AtomicLong(0));
            registerBrokerBody.getTopicConfigSerializeWrapper().getDataVersion().setTimestamp(0L);
            registerBrokerBody.getTopicConfigSerializeWrapper().getDataVersion().setStateVersion(0L);
        }
        return registerBrokerBody;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/8
     * @方法描述：从Namesrv的路由管理器中注销Broker信息的方法
     */
    public RemotingCommand unregisterBroker(ChannelHandlerContext ctx, RemotingCommand request) throws RemotingCommandException {

        //创建响应对象
        final RemotingCommand response = RemotingCommand.createResponseCommand(null);
        //解码请求对象，得到注销Broker信息的请求头
        final UnRegisterBrokerRequestHeader requestHeader = (UnRegisterBrokerRequestHeader) request.decodeCommandCustomHeader(UnRegisterBrokerRequestHeader.class);
        //向路由信息管理器提交注销Broker信息的请求
        if (!this.namesrvController.getRouteInfoManager().submitUnRegisterBrokerRequest(requestHeader)) {
            //如果提交失败了则记录告警信息
            log.warn("Couldn't submit the unregister broker request to handler, broker info: {}", requestHeader);
            //设置系统错误响应码
            response.setCode(ResponseCode.SYSTEM_ERROR);
            response.setRemark(null);
            return response;
        }
        //提交成功则回复客户端成功响应
        response.setCode(ResponseCode.SUCCESS);
        response.setRemark(null);
        return response;
    }


    /**
     * @课程描述:从零带你写框架系列中的课程，整个系列包含netty，xxl-job，rocketmq，nacos，sofajraft，spring，springboot，disruptor，编译器，虚拟机等等。
     * @author：陈清风扬，个人微信号：chenqingfengyangjj。
     * @date:2025/1/14
     * @方法描述：得到Broker集群信息的方法
     */
    private RemotingCommand getBrokerClusterInfo(ChannelHandlerContext ctx, RemotingCommand request) {
        final RemotingCommand response = RemotingCommand.createResponseCommand(null);
        //从路由信息管理器中得到所有Broker集群的信息
        byte[] content = this.namesrvController.getRouteInfoManager().getAllClusterInfo().encode();
        response.setBody(content);
        response.setCode(ResponseCode.SUCCESS);
        response.setRemark(null);
        return response;
    }

    private boolean checksum(ChannelHandlerContext ctx, RemotingCommand request, RegisterBrokerRequestHeader requestHeader) {

        if (requestHeader.getBodyCrc32() != 0) {
            final int crc32 = UtilAll.crc32(request.getBody());
            if (crc32 != requestHeader.getBodyCrc32()) {
                log.warn(String.format("receive registerBroker request,crc32 not match,from %s",
                        RemotingHelper.parseChannelRemoteAddr(ctx.channel())));
                return false;
            }
        }
        return true;
    }
    @Override
    public boolean rejectRequest() {
        return false;
    }


}